package com.atguigu.flink.edu.app.fun;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.flink.edu.beans.TableProcess;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.*;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class BaseDbTableProcessFunction extends BroadcastProcessFunction<JSONObject,String, JSONObject> {
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;

    public BaseDbTableProcessFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }
    private Map<String,TableProcess> configMap = new HashMap<>();
    @Override
    public void open(Configuration parameters) throws Exception {
        //注册驱动
        Class.forName("com.mysql.cj.jdbc.Driver");
        //建立连接
        Connection conn = DriverManager.getConnection("jdbc:mysql://hadoop001:3306/edu_config?" +
                "user=root&password=000000&useUnicode=true&characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false");
        String sql = "select * from edu_config.table_process where sink_type = 'dwd'";
        //创建数据库操作对象
        PreparedStatement prepareStatement = conn.prepareStatement(sql);
        //执行sql
        ResultSet ps = prepareStatement.executeQuery();
        //处理结果集
        ResultSetMetaData metaData = ps.getMetaData();
        while (ps.next()){
            JSONObject jsonObject = new JSONObject();
            for (int i = 1; i <= metaData.getColumnCount(); i++) {
                String columnName = metaData.getColumnName(i);
                Object psObject = ps.getObject(i);
                jsonObject.put(columnName,psObject);
            }
            //将json对象转换为Tableprocess
            TableProcess tableProcess = jsonObject.toJavaObject(TableProcess.class);
            //将读取到的配置信息放到map集合中
            String sourceTable = tableProcess.getSourceTable();
            String sourceType = tableProcess.getSourceType();
            String key = sourceTable + "_" + sourceType;
            configMap.put(key,tableProcess);
        }
        conn.close();
        ps.close();
        prepareStatement.close();
    }

    @Override
    public void processElement(JSONObject jsonObject, ReadOnlyContext readOnlyContext, Collector<JSONObject> out) throws Exception {
        String table = jsonObject.getString("table");
        String type = jsonObject.getString("type");
        String key = table + "_" + type;
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = readOnlyContext.getBroadcastState(mapStateDescriptor);
        TableProcess tableProcess = null;
        if ((tableProcess = broadcastState.get(key))!=null || (tableProcess = configMap.get(key))!=null ){
            JSONObject dataJsonObj = jsonObject.getJSONObject("data");
            filterColumn(dataJsonObj,tableProcess.getSinkColumns());
            Long ts = jsonObject.getLong("ts");
            //添加时间字段
            dataJsonObj.put("ts",ts);
            //添加表字段
            dataJsonObj.put("sink_table",tableProcess.getSinkTable());
            out.collect(dataJsonObj);
        }
    }

    private void filterColumn(JSONObject dataJsonObj, String sinkColumns) {
        String[] split = sinkColumns.split(",");
        List<String> list = Arrays.asList(split);
        dataJsonObj.entrySet().removeIf(entry -> !list.contains(entry.getKey()));
    }

    @Override
    public void processBroadcastElement(String sqlDs, Context context, Collector<JSONObject> out) throws Exception {
        BroadcastState<String, TableProcess> broadcastState = context.getBroadcastState(mapStateDescriptor);
        JSONObject jsonObjSqlDs = JSON.parseObject(sqlDs);
        //获取配置表中的操作类型'op'
        String op = jsonObjSqlDs.getString("op");
        if ("d".equals(op)){
            TableProcess before = jsonObjSqlDs.getObject("before", TableProcess.class);
            String sinkType = before.getSinkType();
            if ("dwd".equals(sinkType)){
                String sourceTable = before.getSourceTable();
                String sourceType = before.getSourceType();
                String key = sourceTable + "_" + sourceType;
                broadcastState.remove(key);
                configMap.remove(key);
            }
        }else {
            TableProcess after = jsonObjSqlDs.getObject("after", TableProcess.class);
            String sinkType = after.getSinkType();
            if ("dwd".equals(sinkType)){
                String sourceTable = after.getSourceTable();
                String sourceType = after.getSourceType();
                String key = sourceTable + "_" + sourceType;
                broadcastState.put(key,after);
                configMap.put(key,after);
            }
        }
    }
}
//处理广播流数据
//    //"op":"r": {"before":null,"after":{"source_table":"activity_info","source_type":"ALL","sink_table":"dim_activity_info","sink_type":"dim","sink_columns":"id,activity_name,activity_type,activity_desc,start_time,end_time,create_time","sink_pk":"id","sink_extend":null},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":0,"snapshot":"false","db":"gmall0905_config","sequence":null,"table":"table_process","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1676877345748,"transaction":null}
//    //"op":"c": {"before":null,"after":{"source_table":"a","source_type":"ALL","sink_table":"aa","sink_type":"dim","sink_columns":"id,name","sink_pk":"id","sink_extend":null},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1676877468000,"snapshot":"false","db":"gmall0905_config","sequence":null,"table":"table_process","server_id":1,"gtid":null,"file":"mysql-bin.000011","pos":727651,"row":0,"thread":null,"query":null},"op":"c","ts_ms":1676877468250,"transaction":null}
//    //"op":"u": {"before":{"source_table":"a","source_type":"ALL","sink_table":"aa","sink_type":"dim","sink_columns":"id,name","sink_pk":"id","sink_extend":null},"after":{"source_table":"a","source_type":"ALL","sink_table":"aa","sink_type":"dim","sink_columns":"id,name,age","sink_pk":"id","sink_extend":null},"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1676877507000,"snapshot":"false","db":"gmall0905_config","sequence":null,"table":"table_process","server_id":1,"gtid":null,"file":"mysql-bin.000011","pos":727985,"row":0,"thread":null,"query":null},"op":"u","ts_ms":1676877507224,"transaction":null}
//    //"op":"d": {"before":{"source_table":"a","source_type":"ALL","sink_table":"aa","sink_type":"dim","sink_columns":"id,name,age","sink_pk":"id","sink_extend":null},"after":null,"source":{"version":"1.6.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1676877536000,"snapshot":"false","db":"gmall0905_config","sequence":null,"table":"table_process","server_id":1,"gtid":null,"file":"mysql-bin.000011","pos":728355,"row":0,"thread":null,"query":null},"op":"d","ts_ms":1676877536960,"transaction":null}

